package nsqlookupd

import (
	"bufio"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"strings"
	"sync/atomic"
	"time"

	"github.com/nsqio/nsq/internal/protocol"
	"github.com/nsqio/nsq/internal/version"
)

type LookupProtocolV1 struct {
	ctx *Context
}
//实现internal/Protocol.go中定义的Protocol的接口的IOLoop方法
func (p *LookupProtocolV1) IOLoop(conn net.Conn) error {
	var err error
	var line string

	client := NewClientV1(conn)
	//此处需要注意为何NewReader可以传入client作为参数。
	//打开client_v1.go可以看到，其中嵌入了net.Conn，用JAVA的思想可以说，ClientV1是继承自net.Conn的。
	//那接下来的问题是：查官方文档http://golang.org/pkg/bufio/#NewReader
	//NewReader的参数类型为io.Reader，这和net.Conn也不同
	//为一探究竟，打开go的源码。分别打开go源码下src\pkg\io\io.go和src\pkg\net\net.go
	//发现io.Reader是一个接口，其中有一个方法 Read(p []byte) (n int, err error)
	//net.Conn也是一个接口，下面有很多方法，其中一个是 Read(b []byte) (n int, err error)
	//可以看出，这两个方法的参数是完全一样的。即net.Conn里的方法完全能覆盖io.Reader里定义的方法
	//这就是我们所喜欢Go的 组合模式
	//所以我们这里可以传入client作为参数
	reader := bufio.NewReader(client)

	for {
		// 读取一行数据
		line, err = reader.ReadString('\n')
		if err != nil {
			break
		}
		// 去掉两边的空格
		line = strings.TrimSpace(line)
		//将数据用空格分割成数组，根据后面的代码可看出，第一个参数是动作类型，包括四种：PING IDENTIFY REGISTER UNREGISTER
		params := strings.Split(line, " ")

		var response []byte
		//调用LookupProtocolV1的Exec方法
		response, err = p.Exec(client, reader, params)

		if err != nil {
			ctx := ""
			if parentErr := err.(protocol.ChildErr).Parent(); parentErr != nil {
				ctx = " - " + parentErr.Error()
			}
			p.ctx.nsqlookupd.logf(LOG_ERROR, "[%s] - %s%s", client, err, ctx)

			//返回错误给客户端，SendResponse方法在internal/Protocol.go中定义
			_, sendErr := protocol.SendResponse(client, []byte(err.Error()))

			if sendErr != nil {
				p.ctx.nsqlookupd.logf(LOG_ERROR, "[%s] - %s%s", client, sendErr, ctx)
				break
			}
			// errors of type FatalClientErr should forceably close the connection
			if _, ok := err.(*protocol.FatalClientErr); ok {
				break
			}
			continue
		}
		//Exec方法返回了响应数据，将响应发送到客户端
		if response != nil {
			_, err = protocol.SendResponse(client, response)
			if err != nil {
				break
			}
		}
	}
	//如果前面的for循环退出了，则表示程序要退出了，将注册信息都从RegistrationDB中删除
	conn.Close()
	p.ctx.nsqlookupd.logf(LOG_INFO, "CLIENT(%s): closing", client)
	if client.peerInfo != nil {
		registrations := p.ctx.nsqlookupd.DB.LookupRegistrations(client.peerInfo.id)
		for _, r := range registrations {
			if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {
				p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
					client, r.Category, r.Key, r.SubKey)
			}
		}
	}
	return err
}
//请求分发，根据每行数据的第一个参数，调用不同的方法
func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
	switch params[0] {
	case "PING":
		return p.PING(client, params)
	case "IDENTIFY":
		return p.IDENTIFY(client, reader, params[1:])
	case "REGISTER":
		return p.REGISTER(client, reader, params[1:])
	case "UNREGISTER":
		return p.UNREGISTER(client, reader, params[1:])
	}
	//数据不合法，提示出错。
	return nil, protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))
}

func getTopicChan(command string, params []string) (string, string, error) {
	if len(params) == 0 {
		return "", "", protocol.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("%s insufficient number of params", command))
	}

	topicName := params[0]
	var channelName string
	if len(params) >= 2 {
		channelName = params[1]
	}

	if !protocol.IsValidTopicName(topicName) {
		return "", "", protocol.NewFatalClientErr(nil, "E_BAD_TOPIC", fmt.Sprintf("%s topic name '%s' is not valid", command, topicName))
	}

	if channelName != "" && !protocol.IsValidChannelName(channelName) {
		return "", "", protocol.NewFatalClientErr(nil, "E_BAD_CHANNEL", fmt.Sprintf("%s channel name '%s' is not valid", command, channelName))
	}

	return topicName, channelName, nil
}
// 根据传进来的param 注册一个 topic或者channel
func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
	if client.peerInfo == nil {
		return nil, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
	}
	//调用本文件里的getTopicChan方法,从参数params中取的topic和channel
	topic, channel, err := getTopicChan("REGISTER", params)
	if err != nil {
		return nil, err
	}

	if channel != "" {
		//定义Registration类型的变量，category为channel,Key为topic,SubKey为channel
		key := Registration{"channel", topic, channel}
		//将client做为一个producer保存在RegistrationDB中
		if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {
			p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) REGISTER category:%s key:%s subkey:%s",
				client, "channel", topic, channel)
		}
	}
	//定义Registration类型的变量，category为topic,Key为topic,SubKey为空
	key := Registration{"topic", topic, ""}
	if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {
		p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) REGISTER category:%s key:%s subkey:%s",
			client, "topic", topic, "")
	}

	return []byte("OK"), nil
}
// 删除topic或者channel
func (p *LookupProtocolV1) UNREGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
	if client.peerInfo == nil {
		return nil, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
	}

	topic, channel, err := getTopicChan("UNREGISTER", params)
	if err != nil {
		return nil, err
	}
	//params参数中有指定channel
	if channel != "" {
		key := Registration{"channel", topic, channel}
		removed, left := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id)
		if removed {
			p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
				client, "channel", topic, channel)
		}
		// for ephemeral channels, remove the channel as well if it has no producers
		//把没有producers并标注为ephemeral（中文意思为短暂的）的channels也删除了，此句话翻译了作者的注释
		//但是看起来有些不太懂，主要原因是目前对channel topic的概念还不清楚，这个也需待后续明了，我估计是有临时channel这玩意儿。
		//left=0表示这个Registration下面所有的Producer都被删完了
		if left == 0 && strings.HasSuffix(channel, "#ephemeral") {
			p.ctx.nsqlookupd.DB.RemoveRegistration(key)
		}
	} else {
		// no channel was specified so this is a topic unregistration
		// remove all of the channel registrations...
		// normally this shouldn't happen which is why we print a warning message
		// if anything is actually removed
		//没有指定channel，所以是topic的移除
		//删除掉所有channel的注册
		//一般来说，这不应该发生，如果确实有东西被删掉了，是不正常的，所以在下面的LOG里打印了一个warning
		registrations := p.ctx.nsqlookupd.DB.FindRegistrations("channel", topic, "*")
		for _, r := range registrations {
			if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {
				p.ctx.nsqlookupd.logf(LOG_WARN, "client(%s) unexpected UNREGISTER category:%s key:%s subkey:%s",
					client, "channel", topic, r.SubKey)
			}
		}

		key := Registration{"topic", topic, ""}
		if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id); removed {
			p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",
				client, "topic", topic, "")
		}
	}

	return []byte("OK"), nil
}
// Client与nsqdlookupd连接后，在进行PING REGISTER UNREGISTER操作之前，必须先IDENTIFY,通过IDENTIFY来初始化peerInfo。
func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {
	var err error
	//不能重复初始化peerInfo，已经初始化时，返回错误
	if client.peerInfo != nil {
		return nil, protocol.NewFatalClientErr(err, "E_INVALID", "cannot IDENTIFY again")
	}
	//读取数据的长度到bodyLen变量中，用于判断数据包结尾
	var bodyLen int32
	err = binary.Read(reader, binary.BigEndian, &bodyLen)
	if err != nil {
		return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body size")
	}
	//取bodyLen长度的数据。
	body := make([]byte, bodyLen)
	_, err = io.ReadFull(reader, body)
	if err != nil {
		return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body")
	}
	// body is a json structure with producer information
	//body是一个json结构的数据
	//PeerInfo类型在nsqlookupd\registration_db.go文件中定义
	//PeerInfo类型在Producer类型和ClientV1类型中都有使用
	//id作为PeerInfo的唯一性标识，将在后续经常使用
	// 这个id为远程地址
	peerInfo := PeerInfo{id: client.RemoteAddr().String()}
	err = json.Unmarshal(body, &peerInfo)
	if err != nil {
		return nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to decode JSON body")
	}

	peerInfo.RemoteAddress = client.RemoteAddr().String()

	// require all fields
	//校验JSON传来的数据是否完整
	if peerInfo.BroadcastAddress == "" || peerInfo.TCPPort == 0 || peerInfo.HTTPPort == 0 || peerInfo.Version == "" {
		return nil, protocol.NewFatalClientErr(nil, "E_BAD_BODY", "IDENTIFY missing fields")
	}

	//修改peerInfo的lastUpdate值为当前时间
	atomic.StoreInt64(&peerInfo.lastUpdate, time.Now().UnixNano())

	p.ctx.nsqlookupd.logf(LOG_INFO, "CLIENT(%s): IDENTIFY Address:%s TCP:%d HTTP:%d Version:%s",
		client, peerInfo.BroadcastAddress, peerInfo.TCPPort, peerInfo.HTTPPort, peerInfo.Version)

	//把当前client加入到RegistrationDB的记录里。Registration的category是"client"，Key和SubKey为空
	client.peerInfo = &peerInfo
	if p.ctx.nsqlookupd.DB.AddProducer(Registration{"client", "", ""}, &Producer{peerInfo: client.peerInfo}) {
		p.ctx.nsqlookupd.logf(LOG_INFO, "DB: client(%s) REGISTER category:%s key:%s subkey:%s", client, "client", "", "")
	}

	// build a response
	//构建一个响应给client
	data := make(map[string]interface{})
	//返回nsqlookupd监听的TCP端口
	data["tcp_port"] = p.ctx.nsqlookupd.RealTCPAddr().Port
	//返回nsqlookupd监听的HTTP端口
	data["http_port"] = p.ctx.nsqlookupd.RealHTTPAddr().Port
	//nsq版本
	data["version"] = version.Binary
	hostname, err := os.Hostname()
	if err != nil {
		log.Fatalf("ERROR: unable to get hostname %s", err)
	}
	data["broadcast_address"] = p.ctx.nsqlookupd.opts.BroadcastAddress
	data["hostname"] = hostname

	//转化为一个JSON字符串
	response, err := json.Marshal(data)
	if err != nil {
		p.ctx.nsqlookupd.logf(LOG_ERROR, "marshaling %v", data)
		return []byte("OK"), nil
	}
	return response, nil
}

func (p *LookupProtocolV1) PING(client *ClientV1, params []string) ([]byte, error) {
	if client.peerInfo != nil {
		// we could get a PING before other commands on the same client connection
		//我们可以在同一个客户端连接上获得一个PING信号
		cur := time.Unix(0, atomic.LoadInt64(&client.peerInfo.lastUpdate))
		now := time.Now()
		p.ctx.nsqlookupd.logf(LOG_INFO, "CLIENT(%s): pinged (last ping %s)", client.peerInfo.id,
			now.Sub(cur))
		//修改client.peerInfo的lastUpdate值为当前时间
		atomic.StoreInt64(&client.peerInfo.lastUpdate, now.UnixNano())
	}
	return []byte("OK"), nil
}
